#ifndef BASIC_ALGORITHMS_PARALLE_SORT_H
#define BASIC_ALGORITHMS_PARALLE_SORT_H

#include <boost/thread.hpp>
#include <algorithm>
#include <string.h>

template<typename T>
struct Range {
  //Range(int *begin, int *end) : begin(begin), end(end) {}
  void Set(T *begin, T *end) { this->begin = begin; this->end = end; }
  T *begin, *end;  
};

template<typename T>
class SortHandle {
public:
  SortHandle(Range<T> &range) : range_(range) { }
  void operator()() {
    std::sort(range_.begin, range_.end);
  }
private:
  Range<T> &range_;
};

template<typename T>
class MergeHandle {
public:
  MergeHandle(Range<T> &range1, Range<T> &range2, Range<T> *result)
    : range1_(range1), range2_(range2), result_(result) { }
  void operator()() {
    T *it1 = range1_.begin;
    T *it2 = range2_.begin;
    T *res_ptr = result_->begin;
    while (it1 != range1_.end && it2 != range2_.end) {
      if (*it1 < *it2) {
        *res_ptr = *it1; ++it1;
      } else {
        *res_ptr = *it2; ++it2;
      }
      ++res_ptr;
    } // end of while
    if (it1 != range1_.end) {
      memcpy(res_ptr, it1, (range1_.end - it1) * sizeof(T));
    } else {
      memcpy(res_ptr, it2, (range2_.end - it2) * sizeof(T));
    }
  }
private:
  Range<T> &range1_, &range2_;
  Range<T> *result_;
};

template<typename T>
void parallel_sort(T data[], int size) {
  boost::thread_group sort_tg;
  T *ptr = data;
  int grain_size = size / 8;
  Range<T> level1_ranges[8];  
  for (int i = 0; i < 7; ++i) {
    level1_ranges[i].Set(ptr, ptr + grain_size);    
    ptr += grain_size;
  }
  level1_ranges[7].Set(ptr, data + size);

  for (int i = 0; i < 8; ++i) {
    sort_tg.create_thread(SortHandle<T>(level1_ranges[i]));
  }  
  sort_tg.join_all();

  //merge  
  T *temp_data = new T[size];
  T *temp_ptr = temp_data;
  Range<T> temp1_ranges[4];
  int grain_size1 = grain_size + grain_size;
  for (int i = 0; i < 3; ++i) {
    temp1_ranges[i].Set(temp_ptr, temp_ptr + grain_size1);
    temp_ptr += grain_size1;
  }
  temp1_ranges[3].Set(temp_ptr, temp_data + size);

  boost::thread_group merge1_tg;  
  for (int i = 0, j = 0; i < 4; ++i, j += 2) {
    merge1_tg.create_thread(MergeHandle<T>(level1_ranges[j], level1_ranges[j+1],
                                        &temp1_ranges[i]));
  }
  merge1_tg.join_all();

  Range<T> temp2_ranges[2];
  int grain_size2 = grain_size1 + grain_size1;
  temp2_ranges[0].Set(data, data + grain_size2);
  temp2_ranges[1].Set(data + grain_size2, data + size);
  
  boost::thread_group merge2_tg;
  for (int i = 0, j = 0; i < 2; ++i, j += 2) {
    merge2_tg.create_thread(MergeHandle<T>(temp1_ranges[j], temp1_ranges[j+1],
                                        &temp2_ranges[i]));
  }
  merge2_tg.join_all();

  Range<T> temp3_range;
  temp3_range.Set(temp_data, temp_data + size);
  MergeHandle<T> merge3(temp2_ranges[0], temp2_ranges[1], &temp3_range);
  merge3();
  memcpy(data, temp_data, size * sizeof(T));
  delete[] temp_data;
}

#endif // BASIC_ALGORITHMS_PARALLE_SORT_H